/**
 * Aggregation2.java
 * ************************** 
 * @date Apr 6, 2011
 * @author Michele Tamburini
 * 
 * **************************
 */
package gossipServices.aggregation;

import gossipServices.aggregation.functions.AggregationUpdateFunction;
import gossipServices.basic.view.PartialView;
import gossipServices.gossipThread.GossipThread;
import gossipServices.gossipThread.GossipThreadsInstructions;
import gossipServices.network.NetworkServicesAPI;
import gossipServices.pss.PeerSamplingServiceAPI;

//import java.util.logging.Level;
//import java.util.logging.Logger;

/**
 * Description: 
 * This class extends GossipThread so its a full working
 * gossip thread. 
 * It encapsulates all the required variables of the 
 * Aggregation protocol and can be used as a base class
 * for further implementations.
 * 
 * A class extending from this one has to implement the 
 * GossipThreadsInstructions interface.
 *
 */
public abstract class AggregationMiddleware extends GossipThread
	implements GossipThreadsInstructions, AggregationService{
	
	protected static boolean AGGREGATION_PRINT_VALUE = false;
		
	/**
	 * Looking in the paper the convergence factor for a view of 
	 * c = 30 takes around 30 cycles.
	 * In this implementation the epochCyclesLenth is fixed,
	 * but more sophisticated ones may use the 'accuracy' value
	 * of the variance in each node.
	 */
	protected static int epochCyclesLength = 50; 
		
	protected static EpochHandler epochHandler = new ConcreteEpochHandler(epochCyclesLength);
		
	protected PeerSamplingServiceAPI pss;
	protected NetworkServicesAPI nia;
	protected AggregationUpdateFunction updateFunc;

	protected static double startingValue = 0;
	protected static SingleValueContainer estimationValue = new SingleValueContainer();
	protected static SingleValueDouble lastEstimationValue = new SingleValueContainer();
	

//	protected Logger log;
	
	protected static AggregationLogWriter logWriter;
	
	protected boolean activeLogging = true;
	
	/**
	 * @param name
	 * @param cycleLenghtMsec
	 * @param view2
	 */
	protected AggregationMiddleware(String name, long cycleLenghtMsec, PartialView view,
			PeerSamplingServiceAPI pss, NetworkServicesAPI nia, 
			AggregationUpdateFunction function, double startValue) {
		super(name, cycleLenghtMsec, view);
		this.pss = pss;
		this.nia = nia;
		this.updateFunc = function;
		setGossipProtocol(this);		
//		log = Logger.getLogger(name+"Log");	
//		log.setLevel(Level.OFF);
		startingValue = startValue;
		estimationValue.setValue(startValue);
//		log.info("thread: "+currentThread().getName()+" sets value: "+estimationValue);
		
		logWriter = new AggregationLogWriter(nia.getNodeName());
	}
		
	/**
	 * @param sq
	 */
	protected void exchange(double sq) {
		
		synchronized (estimationValue) {
			double sp = estimationValue.getValue();
			// ************************
			// **** UPDATE FUNCTION ***
			estimationValue.setValue((updateFunc.update(sp, sq)));
			// ************************
		}
		
		/*
		 * EPOCH Cycles Handling
		 * suppose that only the complete send + receive with 
		 * a successful exchange leads to an "correct" elapsed cycle
		 */
		try {
			epochHandler.elapsedCycle();
//			log.finer(">>> EPOCH: "+epochHandler.getEpoch()+" cycle: "+epochHandler.getElapsedCycles());		
		} catch (AggregationRestartException e) {
//			e.printStackTrace();
			restart(e);
		}
		
		if(AGGREGATION_PRINT_VALUE)
			System.out.println("Aggregation actual value: "+estimationValue+" epoch "+
				+epochHandler.getEpoch()+" cycle: "+epochHandler.getElapsedCycles());
	}

	/**
	 * @param restartException
	 * @param remainingCycles
	 */
	protected void restart(AggregationRestartExceptionInterface restartException) {
//		int waitCycles = restartException.getRemainingCycles();
		String logLine;
//		log.finer("restarting");
		double lastAggregationValue;
		
		
		if(activeLogging){
			synchronized (logWriter) {
				logLine = nia.getNodeName()+ " epoch: "+epochHandler.getPrecedingEpoch()+
				" cycle: "+epochHandler.getPrecedingElapsedCycles()+
				" value: "+estimationValue;
				if(logWriter.writeLogLine(logLine) == false)
					System.out.println("problem writing log!!!");
			}				
		}
		
		synchronized (estimationValue) {	
			System.out.println("Aggregation at epoch "+epochHandler.getEpoch()+" " +
					"reached value: "+estimationValue);
			lastAggregationValue = estimationValue.getValue();
			estimationValue.setValue(startingValue);
			System.out.println("Aggregation restarting, actual value: "+estimationValue);
		}
		
		synchronized (lastEstimationValue) {
			lastEstimationValue.setValue(lastAggregationValue);
			lastEstimationValue.notify();
		}
		
		nia.clearReceivedMessages();
		/*
		if(waitCycles > 0){
			//should wait until the beginning of the new epoch
			log.info("should wait "+waitCycles * cycleLenght_msec+" msec");
			// wait(waitCycles) || wake up at the first correct exchanging message
			log.info("wake up!!");
		}
		*/
	}

	@Override
	public abstract void gossipThreadInstructions() throws InterruptedException;
	
	public boolean enableAggregationLogging(){
		return (activeLogging = true);
	}
	
	public boolean disableAggregationLogging(){
		return (activeLogging = false);
	}
	
	@Override
	public AggregationUpdateFunction getAggregationFunction(){
		return updateFunc;
	}
	
	@Override
	public SingleValueDouble getLastStableValue(){
		return lastEstimationValue;
	}
	
	@Override
	public SingleValueDouble getActualValue(){ 
		return estimationValue;
	}
	
	@Override
	public Class<?> getServiceClass() {
		return AggregationService.class;
	}
	
		
}
